今天要介紹剩下的5個retriever方法啦!
再依照自己的需求選擇合適的retriever方法!

6. Long-Context Reorder

使用說明:
這個retriever用於在檢索長上下文時避免性能下降，它通過對檢索的文檔進行重新排序，使最相關的文檔排在前面或最後。

程式碼:

from langchain_chains import LLMChain, StuffDocumentsChain
from langchain_community.document_transformers import LongContextReorder
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_openai import OpenAI

# 創建嵌入
embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")

# 文檔列表
texts = ["Basquetball is a great sport.", "The Boston Celtics won the game by 20 points"]

# 創建向量庫retriever
retriever = Chroma.from_texts(texts, embedding=embeddings).as_retriever(search_kwargs={"k": 10})
docs = retriever.invoke("What can you tell me about the Celtics?")

# 重新排序文檔
reordering = LongContextReorder()
reordered_docs = reordering.transform_documents(docs)

這些retriever各有用途，你可以根據實際需求選擇和組合使用它們來提高檢索效果。

7. MultiVector Retriever

概述:
MultiVector Retriever 是一個強大的檢索器，允許為每個文檔存儲多個向量。這在多種使用場景下非常有用，尤其是在需要精確捕捉文檔中多種語義信息時。LangChain 提供了一個基礎的 MultiVectorRetriever，讓這類設置的查詢變得更加簡單。挑戰在於如何為每個文檔創建多個向量，這篇介紹將涵蓋一些常見的方法來創建這些向量並使用 MultiVectorRetriever 進行檢索。

為每個文檔創建多個向量的方法

1. 較小的片段 (Smaller Chunks):

   • 將文檔拆分為更小的片段，並對這些片段進行嵌入。這種方法可以確保嵌入更精確地反映文檔的語義含義，同時保留上下文信息。這種技術在 ParentDocumentRetriever 中有所應用。

2. 摘要 (Summary):

   • 為每個文檔創建摘要，並對這些摘要進行嵌入。這種方法可以更好地提煉出文檔的核心信息，從而提高檢索的準確性。

3. 假設性問題 (Hypothetical Questions):

   • 為每個文檔生成一組假設性問題，這些問題可能是文檔可以回答的問題，並對這些問題進行嵌入。這樣可以讓檢索更加靈活，並且有更多控制檢索結果的可能性。

代碼示例與應用

1. 使用較小的片段 (Smaller Chunks)

這種方法通過將文檔拆分為較小的子片段，然後進行嵌入，以提高語義精確度。

from langchain.retrievers.multi_vector import MultiVectorRetriever
from langchain_chroma import Chroma
from langchain.storage import InMemoryByteStore
from langchain_community.document_loaders import TextLoader
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
import uuid

# 加載文檔
loaders = [
    TextLoader("../../paul_graham_essay.txt"),
    TextLoader("../../state_of_the_union.txt"),
]
docs = []
for loader in loaders:
    docs.extend(loader.load())

# 將文檔拆分為較小的片段
text_splitter = RecursiveCharacterTextSplitter(chunk_size=10000)
docs = text_splitter.split_documents(docs)

# 創建向量存儲
vectorstore = Chroma(
    collection_name="full_documents", embedding_function=OpenAIEmbeddings()
)

# 創建存儲層，用於存儲父文檔
store = InMemoryByteStore()
id_key = "doc_id"

# 創建 MultiVector Retriever
retriever = MultiVectorRetriever(
    vectorstore=vectorstore,
    byte_store=store,
    id_key=id_key,
)

doc_ids = [str(uuid.uuid4()) for _ in docs]

# 將文檔拆分為更小的子片段
child_text_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
sub_docs = []
for i, doc in enumerate(docs):
    _id = doc_ids[i]
    _sub_docs = child_text_splitter.split_documents([doc])
    for _doc in _sub_docs:
        _doc.metadata[id_key] = _id
    sub_docs.extend(_sub_docs)

# 添加子文檔到向量存儲中
retriever.vectorstore.add_documents(sub_docs)
retriever.docstore.mset(list(zip(doc_ids, docs)))

# 使用向量存儲進行相似性檢索
result = retriever.vectorstore.similarity_search("justice breyer")[0]
print(result.page_content)

# 使用 retriever 返回較大的片段
result = retriever.invoke("justice breyer")[0]
print(len(result.page_content))

2. 使用摘要 (Summary)

這種方法通過為每個文檔生成摘要並對其進行嵌入，從而提高檢索的準確性。

from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_chroma import Chroma
from langchain.storage import InMemoryByteStore
import uuid

# 創建生成摘要的鏈
chain = (
    {"doc": lambda x: x.page_content}
    | ChatPromptTemplate.from_template("Summarize the following document:\n\n{doc}")
    | ChatOpenAI(max_retries=0)
    | StrOutputParser()
)

# 生成摘要
summaries = chain.batch(docs, {"max_concurrency": 5})

# 使用 Chroma 創建向量存儲
vectorstore = Chroma(collection_name="summaries", embedding_function=OpenAIEmbeddings())
store = InMemoryByteStore()
id_key = "doc_id"
retriever = MultiVectorRetriever(
    vectorstore=vectorstore,
    byte_store=store,
    id_key=id_key,
)
doc_ids = [str(uuid.uuid4()) for _ in docs]

# 將摘要作為文檔添加到向量存儲中
summary_docs = [
    Document(page_content=s, metadata={id_key: doc_ids[i]})
    for i, s in enumerate(summaries)
]
retriever.vectorstore.add_documents(summary_docs)
retriever.docstore.mset(list(zip(doc_ids, docs)))

# 使用 retriever 進行檢索
result = retriever.invoke("justice breyer")[0]
print(len(result.page_content))

3. 使用假設性問題 (Hypothetical Questions)

這種方法使用大語言模型 (LLM) 生成每個文檔的假設性問題，這些問題被嵌入到向量存儲中，以提高檢索的靈活性。

from langchain.output_parsers.openai_functions import JsonKeyOutputFunctionsParser
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_chroma import Chroma
from langchain.storage import InMemoryByteStore
import uuid

# 定義生成假設性問題的鏈
functions = [
    {
        "name": "hypothetical_questions",
        "description": "Generate hypothetical questions",
        "parameters": {
            "type": "object",
            "properties": {
                "questions": {
                    "type": "array",
                    "items": {"type": "string"},
                },
            },
            "required": ["questions"],
        },
    }
]

chain = (
    {"doc": lambda x: x.page_content}
    | ChatPromptTemplate.from_template(
        "Generate a list of exactly 3 hypothetical questions that the below document could be used to answer:\n\n{doc}"
    )
    | ChatOpenAI(max_retries=0, model="gpt-4").bind(
        functions=functions, function_call={"name": "hypothetical_questions"}
    )
    | JsonKeyOutputFunctionsParser(key_name="questions")
)

# 生成假設性問題
hypothetical_questions = chain.batch(docs, {"max_concurrency": 5})

# 使用 Chroma 創建向量存儲
vectorstore = Chroma(
    collection_name="hypo-questions", embedding_function=OpenAIEmbeddings()
)
store = InMemoryByteStore()
id_key = "doc_id"
retriever = MultiVectorRetriever(
    vectorstore=vectorstore,
    byte_store=store,
    id_key=id_key,
)
doc_ids = [str(uuid.uuid4()) for _ in docs]

# 將假設性問題作為文檔添加到向量存儲中
question_docs = []
for i, question_list in enumerate(hypothetical_questions):
    question_docs.extend(
        [Document(page_content=s, metadata={id_key: doc_ids[i]}) for s in question_list]
    )
retriever.vectorstore.add_documents(question_docs)
retriever.docstore.mset(list(zip(doc_ids, docs)))

# 使用 retriever 進行檢索
result = retriever.invoke("justice breyer")[0]
print(len(result.page_content))

這些示例展示了如何使用 MultiVector Retriever 來提高檢索效果。不同的方法有助於捕捉文檔的不同面向，從而使檢索結果更加豐富和準確。MultiVector Retriever 提供了一個靈活的平台來管理和查詢這些多樣的嵌入，特別適合需要精確語義理解和複雜查詢的應用場景。

8. Parent Document Retriever

概述:
在進行文檔檢索時，經常會遇到兩種相互矛盾的需求：

1. 需要較小的文檔片段，以便嵌入能夠最準確地反映其含義。如果片段太長，嵌入可能會失去意義。
2. 需要足夠長的文檔片段，以便保留每個片段的上下文。

Parent Document Retriever 通過將文檔拆分成較小的數據塊並存儲來解決這一矛盾。在檢索過程中，它首先獲取這些小片段，然後查找這些片段所屬的父文檔 ID，並返回這些較大的文檔。

父文檔 指的是小片段的原始文檔，這可以是整個原始文檔，也可以是一個較大的片段。

使用 Parent Document Retriever 的步驟

1. 設置基本環境

首先，我們需要一些基本設置，包括文檔加載器、文本分割器以及向量存儲。

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_chroma import Chroma
from langchain_community.document_loaders import TextLoader
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 加載文檔
loaders = [
    TextLoader("../../paul_graham_essay.txt"),
    TextLoader("../../state_of_the_union.txt"),
]
docs = []
for loader in loaders:
    docs.extend(loader.load())

2. 全文檢索模式

在這種模式下，我們希望檢索到完整的文檔，因此僅指定用於生成子文檔的分割器。

# 創建子文檔的文本分割器
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)

# 用於索引子片段的向量存儲
vectorstore = Chroma(
    collection_name="full_documents", embedding_function=OpenAIEmbeddings()
)

# 用於存儲父文檔的存儲層
store = InMemoryStore()

# 創建 ParentDocumentRetriever
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
)

# 添加文檔到檢索器中
retriever.add_documents(docs, ids=None)

# 列出存儲的父文檔的鍵
list(store.yield_keys())

在上面的代碼中，我們將兩個文檔添加到檢索器中，並用文本分割器將其拆分為較小的子文檔片段。

3. 執行相似性檢索

接下來，我們使用向量存儲執行相似性檢索，可以看到檢索器返回的小片段。

sub_docs = vectorstore.similarity_search("justice breyer")

print(sub_docs[0].page_content)

此處的檢索將返回與查詢語句最相似的小片段。

4. 檢索父文檔

最後，我們使用 ParentDocumentRetriever 進行檢索，這會返回與子片段對應的父文檔，通常是較大的片段或完整的文檔。

retrieved_docs = retriever.invoke("justice breyer")

print(len(retrieved_docs[0].page_content))

檢索較大片段

有時候，完整文檔可能過於龐大，因此我們希望先將原始文檔分割成較大的片段，然後再分割成較小片段。在檢索時，我們檢索較大的片段，而不是整個文檔。

# 創建父文檔的文本分割器
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)

# 創建子文檔的文本分割器
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)

# 用於索引子片段的向量存儲
vectorstore = Chroma(
    collection_name="split_parents", embedding_function=OpenAIEmbeddings()
)

# 用於存儲父文檔的存儲層
store = InMemoryStore()

# 創建 ParentDocumentRetriever
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

# 添加文檔到檢索器中
retriever.add_documents(docs)

# 列出存儲的父文檔的鍵
len(list(store.yield_keys()))

通過這樣的設置，我們可以同時獲取較小的子片段和較大的父文檔片段，使檢索結果更加靈活和準確。

總結

Parent Document Retriever 提供了一種靈活的檢索方式，能夠在檢索過程中平衡文檔的長度和嵌入的準確性。通過對文檔進行合理的分割和存儲，可以在檢索時既保留文檔的上下文，又能保證嵌入的語義準確性，非常適合需要在多層次上進行語義檢索的應用場景。

9. Self-querying

概述：
Self-querying 檢索器是一種能夠自我查詢的檢索器。具體來說，給定任何自然語言查詢，該檢索器使用查詢構建 LLM（Large Language Model）鏈來編寫結構化查詢，然後將該結構化查詢應用到其底層的 VectorStore 中。這使得檢索器不僅能使用用戶輸入的查詢與存儲文檔的內容進行語義相似性比較，還能從用戶查詢中提取出與存儲文檔元數據相關的過濾條件並執行這些過濾操作。

如何開始使用

為了演示，我們將使用 Chroma 向量存儲。我們創建了一組包含電影摘要的小型文檔集。

注意：Self-querying 檢索器需要安裝 lark 包。

安裝必要的包

%pip install --upgrade --quiet lark langchain-chroma

創建示例文檔

from langchain_chroma import Chroma
from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings

docs = [
    Document(
        page_content="一群科學家復活了恐龍，結果引發混亂",
        metadata={"year": 1993, "rating": 7.7, "genre": "science fiction"},
    ),
    Document(
        page_content="李奧納多·狄卡皮歐在夢中迷失，夢中夢，夢中夢...",
        metadata={"year": 2010, "director": "Christopher Nolan", "rating": 8.2},
    ),
    Document(
        page_content="一位心理學家/偵探在夢中迷失，夢中夢，這個概念後來被《全面啟動》借用了",
        metadata={"year": 2006, "director": "Satoshi Kon", "rating": 8.6},
    ),
    Document(
        page_content="一群普通大小的女人，性格非常善良，一些男人對她們充滿了愛慕",
        metadata={"year": 2019, "director": "Greta Gerwig", "rating": 8.3},
    ),
    Document(
        page_content="玩具們活了過來，並玩得很開心",
        metadata={"year": 1995, "genre": "animated"},
    ),
    Document(
        page_content="三個男人走進了禁區，三個男人走出了禁區",
        metadata={
            "year": 1979,
            "director": "Andrei Tarkovsky",
            "genre": "thriller",
            "rating": 9.9,
        },
    ),
]

vectorstore = Chroma.from_documents(docs, OpenAIEmbeddings())

創建 Self-querying 檢索器

我們需要提前提供一些信息，關於我們文檔支持的元數據字段以及文檔內容的簡短描述。

from langchain.chains.query_constructor.base import AttributeInfo
from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain_openai import ChatOpenAI

metadata_field_info = [
    AttributeInfo(
        name="genre",
        description="電影的類型，例如：['science fiction', 'comedy', 'drama', 'thriller', 'romance', 'action', 'animated']",
        type="string",
    ),
    AttributeInfo(
        name="year",
        description="電影的發行年份",
        type="integer",
    ),
    AttributeInfo(
        name="director",
        description="電影導演的名字",
        type="string",
    ),
    AttributeInfo(
        name="rating", description="電影的評分（1-10）", type="float"
    ),
]

document_content_description = "電影的簡短摘要"
llm = ChatOpenAI(temperature=0)

retriever = SelfQueryRetriever.from_llm(
    llm,
    vectorstore,
    document_content_description,
    metadata_field_info,
)

測試檢索器

現在，我們可以實際嘗試使用這個檢索器！

範例 1：只指定過濾器

retriever.invoke("我想看評分高於8.5的電影")

# 預期返回：
# [Document(page_content='三個男人走進了禁區，三個男人走出了禁區', metadata={'director': 'Andrei Tarkovsky', 'genre': 'thriller', 'rating': 9.9, 'year': 1979}),
#  Document(page_content='一位心理學家/偵探在夢中迷失，夢中夢，這個概念後來被《全面啟動》借用了', metadata={'director': 'Satoshi Kon', 'rating': 8.6, 'year': 2006})]

範例 2：指定查詢和過濾器

retriever.invoke("Greta Gerwig 是否導演過任何關於女性的電影")

# 預期返回：
# [Document(page_content='一群普通大小的女人，性格非常善良，一些男人對她們充滿了愛慕', metadata={'director': 'Greta Gerwig', 'rating': 8.3, 'year': 2019})]

範例 3：指定複合過濾器

retriever.invoke("有沒有評分高於8.5的科幻電影？")

# 預期返回：
# [Document(page_content='一位心理學家/偵探在夢中迷失，夢中夢，這個概念後來被《全面啟動》借用了', metadata={'director': 'Satoshi Kon', 'rating': 8.6, 'year': 2006}),
#  Document(page_content='三個男人走進了禁區，三個男人走出了禁區', metadata={'director': 'Andrei Tarkovsky', 'genre': 'thriller', 'rating': 9.9, 'year': 1979})]

構建自定義 Self-querying 檢索器

我們可以從頭開始構建檢索器，以獲得更多的自定義控制。首先，我們需要創建一個查詢構建鏈，該鏈將用戶查詢轉換為捕捉用戶指定過濾條件的結構化查詢對象。

from langchain.chains.query_constructor.base import (
    StructuredQueryOutputParser,
    get_query_constructor_prompt,
)

prompt = get_query_constructor_prompt(
    document_content_description,
    metadata_field_info,
)
output_parser = StructuredQueryOutputParser.from_components()
query_constructor = prompt | llm | output_parser

測試自定義查詢構建

query_constructor.invoke(
    {
        "query": "有沒有一些90年代由Luc Besson導演的關於計程車司機的科幻電影"
    }
)

# 預期輸出：
# StructuredQuery(query='計程車司機', filter=Operation(operator=<Operator.AND: 'and'>, arguments=[Comparison(comparator=<Comparator.EQ: 'eq'>, attribute='genre', value='science fiction'), Operation(operator=<Operator.AND: 'and'>, arguments=[Comparison(comparator=<Comparator.GTE: 'gte'>, attribute='year', value=1990), Comparison(comparator=<Comparator.LT: 'lt'>, attribute='year', value=2000)]), Comparison(comparator=<Comparator.EQ: 'eq'>, attribute='director', value='Luc Besson')]), limit=None)

結論

Self-querying 檢索器透過查詢構建鏈，讓查詢過程變得更加靈活。它能夠根據用戶的查詢動態構建結構化查詢，不僅能進行語義相似性檢索，還能應用元數據過濾器，實現更加精確的檢索結果。這在需要複雜查詢的應用場景中非常實用。

10. Time-weighted Vector Store Retriever

概述：
Time-weighted Vector Store Retriever 是一種結合了語義相似度與時間衰減的檢索器。這種檢索器不僅根據文本的語義相似度進行檢索，還根據文檔上次訪問時間與當前時間的間隔來調整檢索結果的權重。通過這種方式，最近頻繁被訪問的文檔將會獲得更高的權重，即被認為“更新鮮”。

算法介紹

該檢索器使用的打分算法為：

[
\text{score} = \text{semantic_similarity} + (1.0 - \text{decay_rate}) ^ \text{hours_passed}
]

這裡的 hours_passed 是指該對象在檢索器中上次訪問後經過的小時數，而不是自創建以來的時間。這意味著經常被訪問的對象將保持“新鮮”。

使用範例

以下是如何使用 Time-weighted Vector Store Retriever 的範例。

初始化

首先，我們定義一個嵌入模型並初始化一個空的向量存儲。

from datetime import datetime, timedelta
import faiss
from langchain.retrievers import TimeWeightedVectorStoreRetriever
from langchain_community.docstore import InMemoryDocstore
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings

# 定義嵌入模型
embeddings_model = OpenAIEmbeddings()

# 初始化向量存儲
embedding_size = 1536
index = faiss.IndexFlatL2(embedding_size)
vectorstore = FAISS(embeddings_model, index, InMemoryDocstore({}), {})

# 創建 Time-weighted Vector Store Retriever
retriever = TimeWeightedVectorStoreRetriever(
    vectorstore=vectorstore, decay_rate=0.0000000000000000000000001, k=1
)

添加文檔並進行檢索

假設我們有兩個文檔 "hello world" 和 "hello foo"。我們將這些文檔添加到檢索器中，並使用不同的時間戳來模擬時間衰減效果。

yesterday = datetime.now() - timedelta(days=1)
retriever.add_documents(
    [Document(page_content="hello world", metadata={"last_accessed_at": yesterday})]
)
retriever.add_documents([Document(page_content="hello foo")])

在這種情況下，因為 "hello world" 是昨天最後一次訪問的，而我們設定的衰減率非常低，所以 "hello world" 仍然會被視為相對新鮮，並且在檢索時優先返回。

result = retriever.invoke("hello world")
print(result)

# 預期返回：
# [Document(page_content='hello world', metadata={'last_accessed_at': datetime.datetime(2023, 12, 27, 15, 30, 18, 457125), 'created_at': datetime.datetime(2023, 12, 27, 15, 30, 8, 442662), 'buffer_idx': 0})]

不同的衰減率設置

衰減率控制了文檔新鮮度的衰減速度。不同的衰減率會影響文檔的檢索優先級。

低衰減率

設定非常低的衰減率意味著文檔會“被記住”更久。實際上，衰減率為 0 時，意味著文檔永遠不會被遺忘，這使得檢索器的行為類似於純向量查詢。

retriever = TimeWeightedVectorStoreRetriever(
    vectorstore=vectorstore, decay_rate=0.0000000000000000000000001, k=1
)

高衰減率

如果將衰減率設置為非常高的值（例如 0.999），那麼文檔的新鮮度會迅速衰減至 0。這種情況下，檢索器的行為也會更接近純向量查詢，但更傾向於最近訪問的文檔。

retriever = TimeWeightedVectorStoreRetriever(
    vectorstore=vectorstore, decay_rate=0.999, k=1
)

在這種情況下，最近訪問的文檔 "hello foo" 將會優先返回，因為 "hello world" 的新鮮度幾乎為 0。

result = retriever.invoke("hello world")
print(result)

# 預期返回：
# [Document(page_content='hello foo', metadata={'last_accessed_at': datetime.datetime(2023, 12, 27, 15, 30, 50, 57185), 'created_at': datetime.datetime(2023, 12, 27, 15, 30, 44, 720490), 'buffer_idx': 1})]

虛擬時間的使用

在測試過程中，可以使用 LangChain 提供的工具來模擬時間。這使得我們可以控制檢索過程中的時間因素，對時間衰減的影響進行測試。

from langchain.utils import mock_now

with mock_now(datetime(2024, 2, 3, 10, 11)):
    result = retriever.invoke("hello world")
    print(result)

結論

Time-weighted Vector Store Retriever 結合了語義相似度和時間衰減的優點，使得它能夠更靈活地處理不同時間點上重要性的變化。通過調整衰減率，可以根據具體應用場景的需求來調整檢索策略，確保最相關和最“新鮮”的信息優先返回。這對於那些需要頻繁更新內容或根據時間進行檢索的應用場景特別有用。

以上就是Retriever的介紹，接下來就要真正進入RAG環節囉!